package com.chenqq.qsocket.mina;

import android.util.Log;

import com.chenqq.qsocket.listener.SocketConnectListener;

import org.apache.mina.core.filterchain.IoFilter;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.ExpiringSessionRecycler;
import org.apache.mina.transport.socket.DatagramSessionConfig;
import org.apache.mina.transport.socket.nio.NioDatagramAcceptor;

import java.net.InetSocketAddress;

import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.BehaviorSubject;

/**
 * UDPService 只接受字符串
 */
public class MinaUSocketService {
    private NioDatagramAcceptor acceptor;
    private InetSocketAddress inetSocketAddress;
    private String tag = MinaSocketService.class.getSimpleName();
    private Disposable subscribeUdpService;
    private SocketConnectListener connectListener;
    public MinaUSocketService() {
        acceptor=new NioDatagramAcceptor();
    }

    public MinaUSocketService setHandler(IoHandlerAdapter ioHandlerAdapter) {
        acceptor.setHandler(ioHandlerAdapter);
        return this;
    }

    public MinaUSocketService setFilterChain(String name, IoFilter filter) {
        acceptor.getFilterChain().addLast(name, filter);
        return this;
    }

    public MinaUSocketService setSessionRecycler(int time){
        acceptor.setSessionRecycler(new ExpiringSessionRecycler(time));
        return this;
    }

    public MinaUSocketService setSendBufferSize(int size){
        DatagramSessionConfig dcfg = acceptor.getSessionConfig();
        dcfg.setSendBufferSize(size);
        return this;
    }

    public MinaUSocketService setReceiveBuffSize(int size){
        DatagramSessionConfig dcfg = acceptor.getSessionConfig();
        dcfg.setReceiveBufferSize(size);
        return this;
    }

    public MinaUSocketService setReuseAddress(boolean isReuse){
        DatagramSessionConfig dcfg = acceptor.getSessionConfig();
        dcfg.setReuseAddress(isReuse);
        return this;
    }

    public MinaUSocketService setMinaPort(int port) {
        inetSocketAddress = new InetSocketAddress(port);
        return this;
    }

    public MinaUSocketService setConnectListener(SocketConnectListener connectListener) {
        this.connectListener = connectListener;
        return this;
    }

    public MinaUSocketService setTag(String tag) {
        this.tag = tag;
        return this;
    }

    public void start() {
        if (subscribeUdpService == null) {
            subscribeUdpService = BehaviorSubject.create(new ObservableOnSubscribe<Boolean>() {
                @Override
                public void subscribe(@NonNull ObservableEmitter<Boolean> emitter) throws Exception {
                    emitter.onNext(true);
                }
            }).unsubscribeOn(Schedulers.io())
                    .subscribeOn(Schedulers.io())
                    .map(new Function<Boolean, String>() {
                        @Override
                        public String apply(@NonNull Boolean o) throws Exception {
                            if (acceptor == null) {
                                return "没有初始化UDP";
                            }
                            if (inetSocketAddress == null) {
                                return "没有初始化UDP端口";
                            }
                            acceptor.bind(inetSocketAddress);
                            return "Success:" + inetSocketAddress.toString();
                        }
                    }).observeOn(Schedulers.io())
                    .subscribe(new Consumer<String>() {
                        @Override
                        public void accept(String s) throws Exception {
                            Log.i("" + tag, s);
                            if (!s.startsWith("Success")) {
                                if (connectListener != null) {
                                    connectListener.isConnect();
                                }
                                stop();
                            } else {
                                if (connectListener != null) {
                                    connectListener.connectFailed(tag, s);
                                }
                            }

                        }
                    }, new Consumer<Throwable>() {
                        @Override
                        public void accept(Throwable throwable) throws Exception {
                            if (connectListener != null) {
                                connectListener.connectFailed(tag, throwable.toString());
                            }
                            stop();
                        }
                    });
        }

    }

    public void stop() {
        if (acceptor != null) {
            acceptor.unbind();
            acceptor.dispose();
            acceptor = null;
        }
        if (subscribeUdpService != null) {
            subscribeUdpService.dispose();
            subscribeUdpService = null;
        }
    }
}
